package com.zh.hudi

import java.time.format.DateTimeFormatter
import java.time.{LocalDateTime, ZoneOffset}
import java.util.concurrent.TimeUnit
import java.util.{Properties, UUID}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.sql.catalyst.dsl.expressions.StringToAttributeConversionHelper

object SendKafkaMessageStreaming {
  val dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")

  val topic = "dev_test_hudi_1026"
  private val brokers = "dev-server001:9092,dev-server002:9092,dev-server003:9092"

  def main(args: Array[String]): Unit = {
    //准备配置属性
    val props = new Properties
    props.put("bootstrap.servers", brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    val producer = new KafkaProducer[String, String](props)

    while (true) {
      val outJson = dataOut
      println(outJson)
      producer.send(new ProducerRecord[String, String](topic, outJson))
      TimeUnit.SECONDS.sleep(args(0).toInt)
    }
    producer.close()
  }

  def dataOut: String = {
    val now = LocalDateTime.now()
    val lc = randomInt(0, 2, 1) match {
      case "0" => "CH"
      case _ => "EN"
    }

    val curd = randomInt(0, 10, 1).toInt match {
      case x if (0 <= x && x < 7) => "INSERT"
      case x if (7 <= x && x < 9) => "UPDATE"
      case 9 => "DELETE"
      case _ => "INSERT"
    }


    val outJson: String =
      $"""
         {
           "data": [
             {
               "id": "${randomInt(0, 10, 3)}",
               "OrderId": "${randomInt(0, 10, 5)}",
               "TrackingIndex": "${randomInt(0, 10, 1)}",
               "TrackingDate": "${dtf.format(now)}",
               "TrackingContent": "${UUID.randomUUID.toString}",
               "TrackingLocation": "${UUID.randomUUID.toString}",
               "TrackingStatus": "${randomInt(0, 10, 2)}",
               "LanguageCode": "${lc}",
               "TrackingSource": "${randomInt(0, 10, 1)}",
               "CreatedOn": "${dtf.format(now)}"
             }
           ],
           "database": "tis_v2_db",
           "es": 1603776222000,
           "id": 5580577,
           "isDdl": false,
           "mysqlType": {
             "id": "bigint(20)",
             "OrderId": "bigint(20)",
             "TrackingIndex": "int(11)",
             "TrackingDate": "datetime",
             "TrackingContent": "varchar(1500)",
             "TrackingLocation": "varchar(100)",
             "TrackingStatus": "int(11)",
             "LanguageCode": "char(2)",
             "TrackingSource": "int(11)",
             "CreatedOn": "datetime"
           },
           "old": null,
           "pkNames": [
             "id"
           ],
           "sql": "",
           "sqlType": {
             "id": -5,
             "OrderId": -5,
             "TrackingIndex": 4,
             "TrackingDate": 93,
             "TrackingContent": 12,
             "TrackingLocation": 12,
             "TrackingStatus": 4,
             "LanguageCode": 1,
             "TrackingSource": 4,
             "CreatedOn": 93
           },
           "table": "trackingorderdetails_3",
           "ts": "${LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli}",
           "type": "${curd}"
         }
         """.toString

    outJson.substring(1)
  }


  def randomInt(min: Int, max: Int, len: Int): String = {
    val sb = new StringBuilder
    for (i <- 0 until len)
      sb.append((Math.random * (max - min + 1) + min).toInt.toString)
    sb.toString
  }

}
